package cn.st4rlight.util.execute;

import static com.github.phantomthief.concurrent.MoreFutures.getUnchecked;
import static java.util.concurrent.CompletableFuture.supplyAsync;

import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.function.Function;

import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.util.concurrent.ThreadFactoryBuilder;

import cn.st4rlight.util.collection.TransformUtil;
import cn.st4rlight.util.exception.ServiceException;
import cn.st4rlight.util.value.NumberUtil;
import lombok.AccessLevel;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;

/**
 * @author st4rlight <st4rlight@163.com>
 * Created on 2024-05-04
 */
@Slf4j
@NoArgsConstructor(access = AccessLevel.PRIVATE)
public class ConcurrentUtil {

    // 线程池名 --> 线程池实例
    private static final Map<String, ExecutorService> THREAD_POOL_MAP = Maps.newConcurrentMap();
    // 一些简单的偶发任务的公用线程池
    public static final ExecutorService EASY_TASK_POOL =
            ConcurrentUtil.buildThreadPool(100, "easy-task-pool");


    /**
     * 自定义线程池并发转换操作
     */
    public static <T, R> List<R> parallelMapTo(List<T> originList,
        Function<T, R> mapFunction, ExecutorService executorService, Duration duration
    ) {
        if (CollectionUtils.isEmpty(originList)) {
            return Lists.newArrayList();
        }

        List<CompletableFuture<R>> futureList = TransformUtil.toList(originList,
                item -> supplyAsync(() -> mapFunction.apply(item), executorService));
        try {
            return TransformUtil.toList(futureList, item -> getUnchecked(item, duration));
        } catch (Exception ex) {
            log.error("parallelMapTo error: {}", ex.getMessage(), ex);
            throw ServiceException.of(ex);
        }
    }

    /**
     * 构建自定义线程池
     *
     * @param threads 线程数
     * @param poolName 自定义线程池名称
     * @return 自定义线程池
     */
    public static ExecutorService buildThreadPool(int threads, String poolName) {
        if (!NumberUtil.isPositive(threads)) {
            throw ServiceException.of(String.format("构建线程池失败, 传入的线程数不合法: %d", threads));
        }

        ThreadFactoryBuilder builder = new ThreadFactoryBuilder();
        poolName = StringUtils.defaultIfBlank(poolName, "default-thread-pool");
        ConcurrentUtil.checkThreadPoolName(poolName);
        builder.setNameFormat(poolName + "-%d");

        ExecutorService resultExecutor = Executors.newFixedThreadPool(threads, builder.build());
        THREAD_POOL_MAP.put(poolName, resultExecutor);
        return resultExecutor;
    }

    private static void checkThreadPoolName(String poolName) {
        if (StringUtils.isBlank(poolName)) {
            throw ServiceException.of("线程池名称不能为空");
        }
        if (THREAD_POOL_MAP.containsKey(poolName)) {
            throw ServiceException.of(String.format("线程池名称[%s]已存在", poolName));
        }
    }
}
